/*
 * Copyright 2009 by Konstantin Bulenkov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bulenkov.crawler;

import com.bulenkov.crawler.event.CrawlerJobListener;
import com.bulenkov.crawler.model.CrawlerModel;
import com.bulenkov.crawler.model.SimpleCrawlerModel;
import com.bulenkov.crawler.util.Downloader;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Konstantin Bulenkov (konstantin@bulenkov.com)
 */
public class MultiThreadCrawlerJob implements CrawlerJob {
  private final String name;
  private CrawlerModel model;
  private long workTime = 0;
  private long pausedTime = 0;
  private long startTime = 0;
  private long pauseTime = 0;
  private boolean paused = true;
  private int maxThreads;
  private List<CrawlerThread> threads = new ArrayList<CrawlerThread>();
  private ThreadGroup group;

  public MultiThreadCrawlerJob(String name, int maxThreads) {
    this.maxThreads = (maxThreads < 1) ? 15 : maxThreads;
    this.name = name;
    group = new ThreadGroup(name);
    model = new SimpleCrawlerModel();
  }

  public MultiThreadCrawlerJob(String name) {
    this(name, 15);
  }

  public String getName() {
    return name;
  }

  public void start() {
    paused = false;
    final long curTime = System.currentTimeMillis();
    startTime = curTime;
    if (pauseTime > 0) {
      pausedTime += curTime - pausedTime;
    }
    for (int i = 0; i < maxThreads; i++) {
      final CrawlerThread thread = new CrawlerThread(group, model);
      threads.add(thread);

      thread.start();
    }
    new StatisticWatcher().start();
  }

  public synchronized void pause() {
    paused = true;
    for (CrawlerThread thread : threads) {
      thread.pause();
    }
    workTime +=System.currentTimeMillis() - startTime;
  }

  public long getTimeElapsed() {
    return paused ? workTime : workTime + System.currentTimeMillis() - startTime;
  }

  public boolean accomplished() {
    return group.activeCount() == 0;
  }

  public void addListener(CrawlerJobListener listener) {
  }

  public boolean removeListener(CrawlerJobListener listener) {
    return false;
  }

  public void setModel(CrawlerModel model) {
    this.model = model;
  }

  public CrawlerModel getModel() {
    return model;
  }

  public ThreadGroup getThreadGroup() {
    return group;
  }

  class StatisticWatcher extends Thread {
    public StatisticWatcher() {
      super();
      setName("StaticticWatcher for " + getName());
      setDaemon(true);
    }

    @Override
    public void run() {
      try {
        while (!accomplished()) {
          sleep(5000);
          info();
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    public void info() {
      long bytes = Downloader.getDownloadedSize(MultiThreadCrawlerJob.this);
      String dwn;
      if (bytes < 1000) {
        dwn = String.format("%d bytes", bytes);
      } else if (bytes < 1000000) {
        dwn = String.format("%.2f Kb", (double) bytes / 1000);
      } else if (bytes < 1000000000) {
        dwn = String.format("%.2f Mb", (double) bytes / 1000000);
      } else {
        dwn = String.format("%.2f Gb", (double) bytes / 1000000000);
      }

      final CrawlerModel model = MultiThreadCrawlerJob.this.model;
      final long seek = model.getCurrentIndex();
      final long linksCount = model.getLinksFound().size();
      final double secondsElapsed = ((double) MultiThreadCrawlerJob.this.getTimeElapsed()) / 1000;
      final int activeThreads = group.activeCount();
      
      System.out.printf("links: %d\t\tcurrent: %d\t\tthreads: %d\t\tspeed: %.2f pages/sec\t\ttraffic: %s\n",
          linksCount,
          seek,
          activeThreads,
          ((double)seek) / secondsElapsed,
          dwn);
    }
  }
}
